package com.moli.iov.disruptor;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.dsl.ProducerType;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * <p>
 *
 *  异步 框架
 *
 * </p>
 *
 * @Auther: zql
 * @Date: 2019/1/2 17:49
 */
public abstract class DataPipeline<D> {

    private static final Logger logger = LoggerFactory.getLogger(DataPipeline.class);

    public int DEFAULT_BUFFER_SIZE = 1024;

    private RingBuffer<DataEvent<D>> ringBuffer;

    private WorkerPool<DataEvent<D>> workerPool;

    public int getBufferSize(){
        return DEFAULT_BUFFER_SIZE;
    }

    protected abstract DataEventHandler<D>[] getHandlers();

    protected abstract void onClose() throws Exception;

    public final void init(ProducerType type, WaitStrategy waitStrategy) throws Exception{
        logger.debug("init ringBuffer begin...");

        DataEventHandler<D>[] handlers = getHandlers();
        if(handlers == null || handlers.length <= 0){
            logger.error("handler不能为空");
            throw new RuntimeException("handler不能为空");
        }
        int ringSize = (int)Math.pow(2.0D, (int)(Math.log(Math.abs(getBufferSize())) / Math.log(2.0D)));
        logger.debug(String.format("set ringSize:%d,real ringBufferSize:%d", new Object[] { Integer.valueOf(ringSize), Integer.valueOf(getBufferSize()) }));

        //构建ringbuffer对象
        this.ringBuffer = RingBuffer.create(type, new EventFactory<DataEvent<D>>() {
            @Override
            public DataEvent<D> newInstance() {
                return new DataEvent<D>();
            }
        }, ringSize, waitStrategy);

        //设置序号栅栏
        SequenceBarrier sequenceBarrier = this.ringBuffer.newBarrier();

        //设置工作池
        workerPool = new WorkerPool<DataEvent<D>>(ringBuffer, sequenceBarrier, new EventExceptionHandler(), handlers);

        //添加sequences
        this.ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        //启动工作池
        workerPool.start(new ThreadPoolExecutor(handlers.length, Runtime.getRuntime().availableProcessors(),
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>()));
        logger.debug("init DataPipeline end");
    }

    /**
     * 发送数据
     * @param data
     */
    public final void onData(D data){
        long sequence = this.ringBuffer.next();
        try{
            DataEvent<D> dDataEvent = ringBuffer.get(sequence);
            dDataEvent.setValue(data);
        }finally{
            ringBuffer.publish(sequence);
        }
    }

    public final void close()
            throws Exception {
        logger.debug("close DataPipeline begin");
        onClose();
        this.workerPool.halt();

        /*for (DataEventHandler dataEventHandler : getHandlers()) {
            dataEventHandler.close();
        }*/
        logger.debug("close DataPipeline end");
    }

    public void setDEFAULT_BUFFER_SIZE(int DEFAULT_BUFFER_SIZE) {
        this.DEFAULT_BUFFER_SIZE = DEFAULT_BUFFER_SIZE;
    }


    static class EventExceptionHandler implements ExceptionHandler<DataEvent> {

        @Override
        public void handleEventException(Throwable ex, long sequence, DataEvent event) {
            logger.error("异步处理异常", ex);
        }

        @Override
        public void handleOnStartException(Throwable ex) {
            logger.error("disruptor 启动异常", ex);
        }

        @Override
        public void handleOnShutdownException(Throwable ex) {
            logger.error("disruptor 停止异常", ex);
        }
    }
}
